package com.ling.rocketmq.producer;


import com.ling.rocketmq.properties.ProducerProperties;
import com.ling.rocketmq.properties.RocketmqSrvProperties;
import com.ling.rocketmq.util.NameSrvAddrUtil;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Objects;

public class MQProducerFactory {

    private final static Logger log = LoggerFactory.getLogger(MQProducerFactory.class);

    private final RocketmqSrvProperties rocketmqSrvProperties;

    private final ProducerProperties producerProperties;

    private DefaultMQProducer producer;

    public MQProducerFactory(RocketmqSrvProperties rocketmqSrvProperties, ProducerProperties producerProperties) {
        this.rocketmqSrvProperties = rocketmqSrvProperties;
        this.producerProperties = producerProperties;
        start();
    }

    private void start() {
        if (Objects.nonNull(producer)) {
            return;
        }
        producer = new DefaultMQProducer(producerProperties.getGroupName());
        producer.setNamesrvAddr(NameSrvAddrUtil.nameSrvAddrBuild(rocketmqSrvProperties.getNameSrvAddr()));
        producer.setMaxMessageSize(producerProperties.getMaxMessageSize());
        producer.setSendMsgTimeout(producerProperties.getSendMsgTimeout());
        producer.setRetryTimesWhenSendAsyncFailed(producerProperties.getRetryTimesWhenSendAsyncFailed());
        try {
            producer.start();
            log.info("rocketmq producer start success !!!");
        } catch (MQClientException e) {
            log.warn("producer 启动失败", e);
        }
    }

    public DefaultMQProducer getProducer() {
        return this.producer;
    }


}